﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Reflection;
using PServiceBus.Core.Provider;
using PServiceBus.Core.Runtime.Configuration;
using System.Collections;
using System.Xml;
using System.Collections.Concurrent;
using PServiceBus.Core.Manager;
using PServiceBus.Core.Runtime.Messages;
using PServiceBus.Core.Runtime;
using PServiceBus.Core.Interface;
using PServiceBus.Core.Security;
using System.Text.RegularExpressions;
using PServiceBus.Core.Runtime.Topics;
using PServiceBus.Core.Runtime.Extensions;
using PServiceBus.Core.Runtime.Transports;
using System.Threading.Tasks;
using PServiceBus.Core.Logger;
using PServiceBus.Gateway.Interfaces.Exceptions;
using PServiceBus.Gateway.Interfaces.DTO;
using DTO = PServiceBus.Gateway.Interfaces.DTO;

namespace PServiceBus.Gateway.Interfaces {
    public sealed class MessageBus : IMessageBus {

        private static Dictionary<string, SubscriberInfo> _subscribers = new Dictionary<string, SubscriberInfo>();
        private static Dictionary<string, DTO.TopicInfo> _topics = new Dictionary<string, DTO.TopicInfo>();
        private static Regex _methodNameRegex = new Regex(@"<(?<MethodName>.+)>", RegexOptions.Compiled | RegexOptions.IgnoreCase | RegexOptions.Multiline);
        private static readonly IAuthorizationProvider _authorizationProvider;
        private bool _disposed;

        public MessageBus() {
        }

        static MessageBus() {
            _authorizationProvider = SecurityManager.AuthorizationProvider;
            InitSubscriptionManager();
        }

        private string _currentUser = null;
        private string CurrentUser {
            get { return _currentUser ?? (_currentUser = Identity.Name) ?? "N/A"; }
        }

        private void ThrowExceptionIfNotAuthorized(Func<IAuthorizationProvider, bool> func) {
            var methodName = _methodNameRegex.Match(func.Method.Name).Groups["MethodName"].Value;
            if (!func(_authorizationProvider)) throw new ESBException(ExceptionConstants.AuthorizationExceptionStr(CurrentUser, methodName));
        }

        private static void InitSubscriptionManager() {
            var managers = SubscriptionManagerManager.Managers;
            foreach (var manager in managers) manager.Init();
        }


        private static void DisposeSubscriptionManager() {
            var managers = SubscriptionManagerManager.Managers;
            foreach (var manager in managers) manager.Dispose();
        }


        private void DoWork(Action<PServiceBus.Core.Interface.ISubscriptionManager> action) {
            var managers = SubscriptionManagerManager.Managers;
            foreach (var manager in managers) {
                try {
                    action(manager);
                } catch (Exception) {
                    if (ReThrowException) throw;
                }
            }
        }

        #region IMessageBus Members
        public bool ReThrowException { get; set; }

        public ESBIdentity Identity { get; set; }

        public Dictionary<string, string> Headers { get; set; }

        public void CreateSubscriber(string subscriber, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanCreateSubscriber(CurrentUser, headers));
            var sub = ReflectionHelper.New<PServiceBus.Core.Interface.ISubscriber>(s => s.Name = subscriber);
            DoWork(s => s.AddSubscriber(sub));
        }

        public void ChangeSubscriberDurability(string subscriber, bool durable, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanEditSubscriber(CurrentUser, subscriber, headers));
            var sub = ReflectionHelper.New<PServiceBus.Core.Interface.ISubscriber>();
            sub.Name = subscriber;
            sub.Durable = durable;
            DoWork(s => s.UpdateSubscriber(sub));
        }

        public void SubscribeTo(string subscriber, string topicName, string filter, bool needHeader, bool caseSensitive, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanSubscribe(CurrentUser, subscriber, topicName, filter, needHeader, headers));
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, topicName, headers));
            var topic = SelectTopic(topicName, headers);
            DoWork(s => s.SubscribeTo(subscriber, topic.ID, topicName, filter, needHeader, caseSensitive));
        }

        public void UpdateSubscriber(string subscriber, string topicName, string filter, bool needHeader, bool caseSensitive, Dictionary<string, string> headers) {
            var currentSubscriber = SubscriberManager.Current.FirstOrDefault(s => s.Name == subscriber);
            if (currentSubscriber == null)
                throw new Exception(String.Format("Specified subscriber[{0}] does not exist", subscriber));
            ThrowExceptionIfNotAuthorized(auth => auth.CanSubscribe(CurrentUser, subscriber, topicName, filter, needHeader, headers));
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, topicName, headers));
            DoWork(s => s.UpdateSubscriber(subscriber, topicName, filter, needHeader, caseSensitive));
        }

        public void UnSubscribeFrom(string subscriber, string topicName, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanEditSubscriber(CurrentUser, subscriber, headers));
            //ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, topicName, headers));
            var topic = SelectTopic(topicName, headers);
            if (!topic.IsValid) throw new TopicNotRegisteredException(ExceptionConstants.TopicExceptionStr(topic.Name));
            DoWork(s => s.UnSubscribeFrom(subscriber, topic.ID, topicName));
        }

        public void AddTransport(string subscriber, string transportName, ITransport transport, string topicName, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanEditSubscriber(CurrentUser, subscriber, headers));
            var isTopicNameNull = String.IsNullOrWhiteSpace(topicName);
            var topic = !isTopicNameNull ? SelectTopic(topicName, headers) : null;
            if (!isTopicNameNull && !topic.IsValid) 
                throw new TopicNotRegisteredException(ExceptionConstants.TopicExceptionStr(topic.Name));
            var topicID = isTopicNameNull ? Guid.Empty : topic.ID;
            DoWork(s => s.AddTransport(subscriber, transportName, transport, topicID));
        }

        public void DeleteTransport(string subscriber, string transportName, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanEditSubscriber(CurrentUser, subscriber, headers));
            DoWork(s => s.DeleteTransport(subscriber, transportName));
        }

        public SubscriberInfo SelectSubscriber(string name, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectSubscriber(CurrentUser, name, headers));
            var sub = SubscriberManager.Current.FirstOrDefault(s => s.Name == name);
            var subscriber = new SubscriberInfo();
            subscriber.Name = name;
            if (sub == null) {
                subscriber.IsValid = false;
                return subscriber;
            }
            subscriber.ID = sub.ID;
            subscriber.IsValid = true;
            subscriber.Durable = sub.Durable;
            foreach (var transport in sub.Transports) {
                var topic = TopicManager.Current.GetTopic(transport.TopicID);
                var topicName = topic == null ? null : topic.Name;
                subscriber.Transports[transport.Name] = new PServiceBus.Gateway.Interfaces.DTO.TransportInfo() { 
                    TopicName = topicName, Name = transport.Name, Transport = transport.Transport };
            }
            foreach (var subscription in sub.Subscriptions) {
                var topic = TopicManager.Current.GetTopic(subscription.ID);
                if (topic == null) continue;
                var tp = SelectTopic(topic.Name, headers);
                tp.Filter = subscription.MessageFilter;
                tp.NeedHeader = subscription.NeedMessageHeaders;
                tp.CaseSensitive = subscription.CaseSensitive;
                subscriber.Topics[topic.Name] = tp;
            }
            return subscriber;
        }

        public DTO.TopicInfo SelectTopic(string name, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, name, headers));
            if (_topics.ContainsKey(name)) return _topics[name];
            var topic = TopicManager.Current.GetTopic(name);
            var tp = new DTO.TopicInfo();
            tp.Name = name;
            if (topic == null) {
                tp.IsValid = false;
                return tp;
            }
            tp.IsValid = true;
            tp.ID = topic.ID;
            tp.Description = topic.Contract.Description;
            var paramInfos = topic.Contract.ParameterInfos ?? new List<PServiceBus.Core.Runtime.Topics.TopicParameterInfo>();
            foreach (var param in paramInfos) tp.Contract.Add(param.Name, param.Description);
            _topics[name] = tp;
            return tp;
        }

        public void DeleteSubscriber(string name, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanEditSubscriber(CurrentUser, name, headers));
            DoWork(s => s.DeleteSubscriber(name));
        }

        public void RegisterTopic(DTO.TopicInfo topic, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanRegisterTopic(CurrentUser, headers));
            if (_topics.ContainsKey(topic.Name)) throw new TopicAlreadyExistException(String.Format("Topic[{0}] already exists", topic.Name));
            var tp = TopicManager.Current.GetTopic(topic.Name);
            if (tp != null) throw new TopicAlreadyExistException(String.Format("Topic[{0}] already exists", topic.Name));
            var publisherTopic = PServiceBus.Core.Runtime.Topics.TopicInfo.New(topic.Name);
            topic.ID = publisherTopic.ID;
            topic.IsValid = true;
            _topics[topic.Name] = topic;
            publisherTopic.Contract = GetContract(topic);
            TopicManager.Current.Attach(publisherTopic);
            RegisterTopicToSubscriptionManager(topic);
        }

        private void RegisterTopicToSubscriptionManager(DTO.TopicInfo topic) {
            var tp = ReflectionHelper.New<PServiceBus.Core.Interface.ITopic>();
            tp.Name = topic.Name;
            tp.Contract = new Core.Runtime.Topics.TopicContract()
            {
                Description = topic.Name ?? topic.Description, Name = topic.Name,
                ParameterInfos = new List<Core.Runtime.Topics.TopicParameterInfo>()
            };
            foreach (var param in topic.Contract)
                tp.Contract.ParameterInfos.Add(Core.Runtime.Topics.TopicParameterInfo.New(param.Key, param.Value));
            DoWork(s => s.RegisterTopic(tp));
        }

        private PServiceBus.Core.Runtime.Topics.TopicContract GetContract(DTO.TopicInfo topic) {
            var contract = Core.Runtime.Topics.TopicContract.New(topic.Name, topic.Description);
            if (topic.Contract != null)
                foreach (var d in topic.Contract)
                    contract.ParameterInfos.Add(Core.Runtime.Topics.
                        TopicParameterInfo.New(d.Key, d.Value));
            return contract;
        }


        public void PublishTopic(DTO.TopicInfo topic, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, topic.Name, headers));
            var tp = default(ITopic);
            var ex = MethodHelper.Try(() => tp = TopicManager.Current.GetTopic(topic.Name));
            if (ex != null) {
                ESBLogger.Log(ex);
                return;
            }
            if (tp == null) throw new TopicNotRegisteredException(String.Format("Could not find the specified topic[{0}]", topic.Name));
            var paramInfos = tp.Contract.ParameterInfos ?? new List<PServiceBus.Core.Runtime.Topics.TopicParameterInfo>();
            var publishDicts = topic.PublishDicts;
            var publishDict = publishDicts[0];
            var counter = 0;
            foreach (var d in publishDict)
                if (paramInfos.Any(p => p.Name == d.Key)) counter++;

            if (paramInfos.Count == counter) {
                PublishTopicMessage(tp, topic, publishDicts);
            } else throw new ESBException(
                   String.Format("Could not publish topic[{0}] because established contract does not match publish information", topic.Name));
        }

        private void PublishTopicMessageAsync(ITopic tp, DTO.TopicInfo topic, List<Dictionary<string, object>> publishDicts) {
            Task.Factory.StartNew(o =>
            {
                var tuple = o as Tuple<ITopic, DTO.TopicInfo, List<Dictionary<string, object>>>;
                PublishTopicMessage(tuple.Item1, tuple.Item2, tuple.Item3);
            }, new Tuple<ITopic, DTO.TopicInfo, List<Dictionary<string, object>>>(tp, topic, publishDicts))
            .ContinueWith(task => PServiceBus.Core.Logger.ESBLogger.Log(task.Exception)
            , TaskContinuationOptions.OnlyOnFaulted)
            .ContinueWith(_ => { });
        }

        private void PublishTopicMessage(ITopic tp, DTO.TopicInfo topic, List<Dictionary<string, object>> publishDicts) {
            MethodHelper.TryLog(() =>
            {
                var paramInfos = tp.Contract.ParameterInfos ?? new List<PServiceBus.Core.Runtime.Topics.TopicParameterInfo>();
                var msg = PServiceBus.Core.Manager.TopicMessageManager.Create(tp, publishDicts, topic.ExpiresIn, topic.Headers);
                msg.Message.Headers[Constants.ESBTOPIC_CREATEDATE] = msg.CreateDate.ToString("yyyy-MM-ddTHH:mm:ss.fffZ");
                msg.Message.Headers[Constants.ESBTOPIC_MESSAGE_ID] = msg.ID.ToString();
                msg.Message.Headers[Constants.ESBTOPIC_TOPIC_NAME] = tp.Name;
                //If Has Contract information then store topic name in first message for re hydrating type information later
                if (paramInfos.Count > 0)
                    msg.Message.Parameters[0][Constants.ESBTOPIC_TOPIC_NAME] = tp.Name;
                if (PServiceBus.Core.Manager.TopicMessageInspectorManager.Inspect(msg)/*Validate message content*/) {
                    PServiceBus.Core.Manager.TopicMessageEnricherManager.Enrich(ref msg);//Enrich message content
                    if (msg != null) PServiceBus.Core.Manager.TopicMessageManager.Add(msg);
                }
            });
        }

        public void DeleteTopic(string name, Dictionary<string, string> headers) {
            ThrowExceptionIfNotAuthorized(auth => auth.CanSelectTopic(CurrentUser, name, headers));
            var topic = TopicManager.Current.GetTopic(name);
            if (topic == null) throw new TopicNotRegisteredException(String.Format("Could not find the specified topic[{0}]", name));
            var canDeleteTopic = SubscriptionCollection.Instance[topic.ID].Count < 1;
            if (!canDeleteTopic) 
                throw new ESBException(String.Format("Could not delete specified topic[{0}] because topic has one or more subscribers", name));
            TopicManager.Current.Detach(topic);
            //Remove if topic id is a wild card
            var subscriptionCollection = SubscriptionCollection.Instance;
            if (subscriptionCollection.RemoveWildCardTopic(topic.ID))
                subscriptionCollection.Save();
            _topics.Remove(name);
        }

        public IList<DTO.TopicInfo> GetTopics(Dictionary<string, string> headers) {
            IList<DTO.TopicInfo> list = new List<DTO.TopicInfo>();
            var topics = TopicManager.Current.ToList();
            foreach (var topic in topics) {
                if (!_authorizationProvider.CanSelectTopic(CurrentUser, topic.Name, headers)) continue;
                DTO.TopicInfo e = null;
                var paramInfos = topic.Contract.ParameterInfos ?? new List<PServiceBus.Core.Runtime.Topics.TopicParameterInfo>();
                if (_topics.ContainsKey(topic.Name)) e = _topics[topic.Name];
                else {
                    e = new DTO.TopicInfo();
                    e.Name = topic.Name;
                    e.ID = topic.ID;
                    e.IsValid = true;
                    e.Description = topic.Contract.Description;
                    foreach (var param in paramInfos) e.Contract.Add(param.Name, param.Description);
                }
                list.Add(e);
            }
            return list;
        }

        public bool IsClosed {
            get { return _disposed; }
        }
        
        #endregion

        #region IDisposable Members

        public void Dispose() {
            if (_disposed) return;
            _disposed = true;
        }

        #endregion
    }
}
